/*
 * Copyright 2009-2010 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.sample.springbatch.job.service;

import com.sample.springbatch.common.MyException;
import com.sample.springbatch.job.dao.MyJdbcExecutionContextDao;
import com.sample.springbatch.job.dao.MyJdbcJobExecutionDao;
import com.sample.springbatch.job.dao.MyJdbcJobInstanceDao;
import com.sample.springbatch.job.dao.MyJdbcStepExecutionDao;
import com.sample.springbatch.scheduling.repository.QuartzInfoRepository;
import com.sample.springbatch.scheduling.repository.QuartzJobDependRepository;
import com.sample.springbatch.scheduling.repository.QuartzJobInfoRepository;
import com.sample.springbatch.scheduling.service.QuartzMgrService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.*;
import org.springframework.batch.core.configuration.ListableJobLocator;
import org.springframework.batch.core.launch.JobExecutionNotRunningException;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.NoSuchJobException;
import org.springframework.batch.core.launch.NoSuchJobExecutionException;
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.JobRestartException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.batch.operations.JobOperator;
import java.util.Collection;
import java.util.Date;
import java.util.Properties;

/**
 * BatchJob服务类
 * @author zhaowg
 */
@Service
public class BatchJobService {
	public static final Logger logger = LoggerFactory.getLogger(BatchJobService.class);

	private String timeZone = "Asia/Shanghai";

	@Autowired
	JobRepository jobRepository;
	@Autowired
	JobOperator jsrJobOperator;
	@Resource(name = "asyncJobLauncher")
	private JobLauncher asyncJobLauncher;
	@Autowired
	ListableJobLocator jobLocator;
    @Autowired
	MyJdbcJobInstanceDao jobInstanceDao;
    @Autowired
	MyJdbcJobExecutionDao jobExecutionDao;
    @Autowired
	MyJdbcStepExecutionDao stepExecutionDao;
	@Autowired
	MyJdbcExecutionContextDao executionContextDao;
	@Autowired
	QuartzJobInfoRepository quartzJobInfoRepository;
	@Autowired
	QuartzJobDependRepository quartzJobDependRepository;
	@Autowired
	QuartzInfoRepository quartzInfoRepository;
	@Autowired
	QuartzMgrService quartzMgrService;



	/**
	 * 执行任务
	 * @param jobName  任务名称
	 * @param jobParameters  任务参数
	 * @return
	 * @throws NoSuchJobException
	 * @throws JobExecutionAlreadyRunningException
	 * @throws JobRestartException
	 * @throws JobInstanceAlreadyCompleteException
	 * @throws JobParametersInvalidException
     */
	public JobExecution launch(String jobName, JobParameters jobParameters, boolean isVerifyDepend) throws NoSuchJobException,
			JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException,
			JobParametersInvalidException, MyException{

		//验证任务的依赖关系
		boolean pass = true;
		if(isVerifyDepend){
			//TODO  验证Job的依赖关系

		}

		JobExecution jobExecution = null;
		if(jobLocator.getJobNames().contains(jobName)) {
			Job job = jobLocator.getJob(jobName);
			JobExecution lastJobExecution = jobRepository.getLastJobExecution(jobName, jobParameters);
			boolean restart = false;
			if (lastJobExecution != null) {
				BatchStatus status = lastJobExecution.getStatus();
				if (status.isUnsuccessful() && status!=BatchStatus.ABANDONED) {
					restart = true;
				}
			}

			if (job.getJobParametersIncrementer() != null && !restart) {
				jobParameters = job.getJobParametersIncrementer().getNext(jobParameters);
			}

			jobExecution = asyncJobLauncher.run(job, jobParameters);

			if (jobExecution.isRunning()) {
				//activeExecutions.add(jobExecution);
			}
		}
		else {
			if(jsrJobOperator != null) {
				jobExecution = new JobExecution(jsrJobOperator.start(jobName, jobParameters.toProperties()));
			}
			else {
				String error = String.format("Unable to find job %s to launch",String.valueOf(jobName));
				logger.error(error);
				throw new NoSuchJobException(error);
			}
		}

		logger.info(String.format("job %s to launch",String.valueOf(jobName)));
		//return new JobExecutionInfo(jobExecution, TimeZone.getTimeZone(this.timeZone));
		return jobExecution;
	}

	/**
	 * 重新启动执行信息
	 * @param jobExecutionId
	 * @param params
	 * @return
	 * @throws NoSuchJobExecutionException
	 * @throws JobExecutionAlreadyRunningException
	 * @throws JobRestartException
	 * @throws JobInstanceAlreadyCompleteException
	 * @throws NoSuchJobException
	 * @throws JobParametersInvalidException
	 */
	public JobExecution restartJobExecution(Long jobExecutionId, JobParameters params) throws NoSuchJobExecutionException,
			JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException,
			NoSuchJobException, JobParametersInvalidException {

		JobExecution jobExecution = null;
		JobExecution target = getJobExecution(jobExecutionId);
		JobInstance lastInstance = target.getJobInstance();

		if(jobLocator.getJobNames().contains(lastInstance.getJobName())) {
			//重启当前定时任务
			Job job = jobLocator.getJob(lastInstance.getJobName());
			JobParameters jobParameters =  new JobParametersBuilder().addString("sys_time",""+System.currentTimeMillis()).toJobParameters();
			jobExecution = asyncJobLauncher.run(job, jobParameters);
		}
		else {
			if(jsrJobOperator != null) {
				if(params != null) {
					jobExecution = new JobExecution(jsrJobOperator.restart(jobExecutionId, params.toProperties()));
				}
				else {
					jobExecution = new JobExecution(jsrJobOperator.restart(jobExecutionId, new Properties()));
				}
			}
			else {
				String error = String.format("Can't find job associated with job execution id %s to restart",
						String.valueOf(jobExecutionId));
				logger.error(error);
				throw new NoSuchJobException(error);
			}
		}

		logger.info(String.format("job[%s] execution id %s to restart", String.valueOf(jobExecution.getJobInstance().getJobName()), String.valueOf(jobExecutionId)));
		//return new JobExecutionInfo(jobExecution, TimeZone.getTimeZone(this.timeZone));
		return jobExecution;
	}


	/**
	 * 停止所有任务执行
	 * @return
	 */
	public int stopAll() {
		Collection<JobExecution> result = jobExecutionDao.getRunningJobExecutions();
		for (JobExecution jobExecution : result) {
			try {
				//Collection<String> jsrJobNames = getJsrJobNames();
				JobExecution je = getJobExecution(jobExecution.getId());
				if(jsrJobOperator != null){//TODO && jsrJobNames.contains(jobExecution.getJobInstance().getJobName())) {
					jsrJobOperator.stop(je.getId());
				}
				else {
					je.stop();
					jobRepository.update(je);
				}
				logger.info(String.format("job[%s] execution id %s to stop", String.valueOf(je.getJobInstance().getJobName()), String.valueOf(je.getId())));
			}catch (NoSuchJobExecutionException e) {
				logger.error(String.format("Can't find job execution id %s to stop", String.valueOf(jobExecution.getId())));
				//e.printStackTrace();
			}
		}
		return result.size();
	}

	/**
	 * 停止任务执行
	 * @param jobExecutionId
	 * @return
	 * @throws NoSuchJobExecutionException
	 * @throws JobExecutionNotRunningException
	 */
	public JobExecution stop(Long jobExecutionId) throws NoSuchJobExecutionException, JobExecutionNotRunningException {

		JobExecution jobExecution = getJobExecution(jobExecutionId);
		if (!jobExecution.isRunning()) {
			String error = "JobExecution is not running and therefore cannot be stopped";
			logger.error(error);
			throw new JobExecutionNotRunningException(error);
		}
		//Collection<String> jsrJobNames = getJsrJobNames();
		if(jsrJobOperator != null){//TODO && jsrJobNames.contains(jobExecution.getJobInstance().getJobName())) {
			jsrJobOperator.stop(jobExecutionId);
			jobExecution = getJobExecution(jobExecutionId);
		}
		else {
			jobExecution.stop();
			jobRepository.update(jobExecution);
		}
		logger.info(String.format("job[%s] execution id %s to stop", String.valueOf(jobExecution.getJobInstance().getJobName()), String.valueOf(jobExecution.getId())));
		//return new JobExecutionInfo(jobExecution, TimeZone.getTimeZone(this.timeZone));
		return jobExecution;

	}

	/**
	 * 放弃任务执行
	 * @param jobExecutionId
	 * @return
	 * @throws NoSuchJobExecutionException
	 * @throws JobExecutionAlreadyRunningException
	 */
	public JobExecution abandon(Long jobExecutionId) throws NoSuchJobExecutionException,
			JobExecutionAlreadyRunningException {

		JobExecution jobExecution = getJobExecution(jobExecutionId);
		if (jobExecution.getStatus().isLessThan(BatchStatus.STOPPING)) {
			String error = "JobExecution is running or complete and therefore cannot be aborted";
			logger.error(error);
			throw new JobExecutionAlreadyRunningException(error);
		}
		//Collection<String> jsrJobNames = getJsrJobNames();
		JobInstance jobInstance = jobExecution.getJobInstance();
		if(jsrJobOperator != null){//TODO && jsrJobNames.contains(jobInstance.getJobName())) {
			jsrJobOperator.abandon(jobExecutionId);
			jobExecution = getJobExecution(jobExecutionId);
		}
		else {
			jobExecution.upgradeStatus(BatchStatus.ABANDONED);
			jobExecution.setEndTime(new Date());
			jobRepository.update(jobExecution);
		}

		logger.info(String.format("job[%s] execution id %s to abandon", String.valueOf(jobExecution.getJobInstance().getJobName()), String.valueOf(jobExecution.getId())));
		//return new JobExecutionInfo(jobExecution, TimeZone.getTimeZone(this.timeZone));
		return jobExecution;
	}


	private JobExecution getJobExecution(Long jobExecutionId) throws NoSuchJobExecutionException {
		JobExecution jobExecution = jobExecutionDao.getJobExecution(jobExecutionId);
		if (jobExecution == null) {
			throw new NoSuchJobExecutionException("There is no JobExecution with id=" + jobExecutionId);
		}
		jobExecution.setJobInstance(jobInstanceDao.getJobInstance(jobExecution));
		try {
			jobExecution.setExecutionContext(executionContextDao.getExecutionContext(jobExecution));
		}
		catch (Exception e) {
			logger.info("Cannot load execution context for job execution: " + jobExecution);
		}
		//stepExecutionDao.addStepExecutions(jobExecution);
		return jobExecution;
	}


}
